package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo7GroupByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local").setAppName("Demo7GroupByKey")

    val sc = new SparkContext(conf)

    val scoresRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    //将RDD转换成kv格式
    val kvRDD: RDD[(String, Int)] = scoresRDD.map(line => {
      val split = line.split(",")
      val id = split(0)
      val score = split(2).toInt

      //key-value格式
      (id, score)
    })

    /**
      * groupByKey   通过key进行分组，将同一个key的所有value分到同一个组内
      * 只有kv格式的RDD才有
      * groupByKey  会产生shuffle ,需要将相同key的数据拉取到同一个分区处理（同一个会进入到同一个reduce原理一样）
      */
    val groupbyByRDD: RDD[(String, Iterable[Int])] = kvRDD.groupByKey()

    val sumScoreRDD: RDD[(String, Int)] = groupbyByRDD.map(kv => {
      val id = kv._1
      /**
        * 迭代器
        * 1、只能被迭代一次
        * 2、数据不会一次全部加载到内存，而是用一点去磁盘读一点（有缓存）
        */
      //同一个学生的所有分数
      val ss: Iterable[Int] = kv._2

      var sum = 0
      for (s <- ss) {
        sum = sum + s
      }

      (id, sum)
    })


    //sumScoreRDD.foreach(println)


    /**
      * groupBy   需要指定分组的字段
      *
      */

    val groupByRDD = scoresRDD.groupBy(line => {
      val id = line.split(",")(0)
      //以学号进行分组
      id
    })

    val resultRDD = groupByRDD.map(kv => {
      val id = kv._1
      val ss = kv._2

      var sum = 0

      for (s <- ss) {
        sum = sum + s.split(",")(2).toInt
      }

      (id, sum)
    })

    resultRDD.foreach(println)


  }

}
